package tjs.study.notes.dotest.disruptor.parallel.spring;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;

/**
 * Created by Nicky.Tang on 2022/9/16 下午4:18
 *
 * @since 02.12.10
 */
@Component
@Configurable
public class DisruptorAutoConfiguration {

    @Bean
    public RingBuffer<DisruptorNotice> ringBuffer() {
        //1 创建RingBuffer
        RingBuffer<DisruptorNotice> ringBuffer =
                RingBuffer.create(ProducerType.MULTI,
                        new EventFactory<DisruptorNotice>() {
                            public DisruptorNotice newInstance() {
                                return new DisruptorNotice();
                            }
                        },

                        // 预实例化xx个DisruptorNotice（常驻内存中，减少gc影响；缓存行）
                        512,
                        // YieldingWaitStrategy：极低延迟，但是会占用极多CPU
                        // SleepingWaitStrategy需要重置下休眠时间，否则默认200ns，仍然会占用较多的cpu
                        new SleepingWaitStrategy(200, Double.valueOf(0.3 * 1000 * 1000 * 1000).longValue())
                );
        return ringBuffer;
    }


    @Bean
    public WorkerPool workerPool(RingBuffer<DisruptorNotice> ringBuffer) {
        //3 创建含有10个消费者的数组:
        DisruptorConsumer[] disruptorConsumers = new DisruptorConsumer[20];
        for (int i = 0; i < disruptorConsumers.length; i++) {
            disruptorConsumers[i] = new DisruptorConsumer("C" + i);
        }

        //4 构建多消费者工作池
        WorkerPool<DisruptorNotice> workerPool = new WorkerPool<DisruptorNotice>(
                ringBuffer,
                ringBuffer.newBarrier(),//通过ringBuffer 创建一个屏障
                new EventHandlerException(),
                disruptorConsumers);

        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //6 启动workerPool
        workerPool.start(Executors.newFixedThreadPool(20));
        return workerPool;
    }

    public static class EventHandlerException implements ExceptionHandler {

        private static final Logger log = LoggerFactory.getLogger(EventHandlerException.class);

        @Override
        public void handleEventException(Throwable throwable, long sequence, Object event) {
            throwable.fillInStackTrace();
            log.error("process data error sequence ==[{}] event==[{}] ,ex ==[{}]", sequence, event.toString(), throwable.getMessage());
        }

        @Override
        public void handleOnStartException(Throwable throwable) {
            log.error("start disruptor error ==[{}]!", throwable.getMessage());
        }

        @Override
        public void handleOnShutdownException(Throwable throwable) {
            log.error("shutdown disruptor error ==[{}]!", throwable.getMessage());
        }
    }
}
